package org.tao.lightningmq.broker;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.config.ConsumeQueueOffsetLoader;
import org.tao.lightningmq.broker.config.GlobalPropertiesLoader;
import org.tao.lightningmq.broker.config.MqTopicLoader;
import org.tao.lightningmq.broker.core.CommitLogAppendHandler;
import org.tao.lightningmq.broker.core.ConsumeQueueAppendHandler;
import org.tao.lightningmq.broker.core.ConsumeQueueConsumeHandler;
import org.tao.lightningmq.broker.enums.BrokerClusterModeEnum;
import org.tao.lightningmq.broker.event.spi.listenr.BrokerMasterChangeListener;
import org.tao.lightningmq.broker.model.MqTopicModel;
import org.tao.lightningmq.broker.netty.broker.BrokerServer;
import org.tao.lightningmq.broker.slave.SlaveSyncService;
import org.tao.lightningmq.broker.timewheel.TimeWheelModelManager;
import org.tao.lightningmq.common.event.EventBus;

/**
 * Broker的主启动类
 */
public class BrokerStartUp {

    private static final Logger LOGGER = LoggerFactory.getLogger(BrokerStartUp.class);

    private static GlobalPropertiesLoader globalPropertiesLoader;
    private static MqTopicLoader mqTopicLoader;
    private static CommitLogAppendHandler commitLogAppendHandler;
    private static ConsumeQueueOffsetLoader consumeQueueOffsetLoader;
    private static ConsumeQueueConsumeHandler consumeQueueConsumeHandler;
    private static ConsumeQueueAppendHandler consumeQueueAppendHandler;
    private static SlaveSyncService slaveSyncService;

    private static void initProperties() throws Exception {
        LOGGER.info("[BrokerStartUp]类, 初始化工作开始!");
        globalPropertiesLoader = new GlobalPropertiesLoader();
        mqTopicLoader = new MqTopicLoader();
        commitLogAppendHandler = new CommitLogAppendHandler();
        consumeQueueOffsetLoader = new ConsumeQueueOffsetLoader();
        consumeQueueConsumeHandler = new ConsumeQueueConsumeHandler();
        consumeQueueAppendHandler = new ConsumeQueueAppendHandler();


        globalPropertiesLoader.loadProperties();
        mqTopicLoader.loadProperties();
        mqTopicLoader.startRefreshMqTopicInfoTask();
        consumeQueueOffsetLoader.loadProperties();
        consumeQueueOffsetLoader.startRefreshConsumeQueueOffsetTask();

        for (MqTopicModel mqTopicModel : CommonCache.getTopicModelList()) {
            String topic = mqTopicModel.getTopic();
            commitLogAppendHandler.prepareMMapLoading(topic);
            consumeQueueAppendHandler.prepareConsumeQueue(topic);
        }
        CommonCache.setConsumeQueueAppendHandler(consumeQueueAppendHandler);
        CommonCache.setConsumeQueueConsumeHandler(consumeQueueConsumeHandler);
        CommonCache.setCommitLogAppendHandler(commitLogAppendHandler);

        // 初始化时间轮
        TimeWheelModelManager timeWheelModelManager = CommonCache.getTimeWheelModelManager();
        timeWheelModelManager.init(new EventBus("time-wheel-event-bus"));
        timeWheelModelManager.doScanTask();

        LOGGER.info("[BrokerStartUp]类, 初始化工作完成!");
    }

    private static void initNameServerChannel() {
        CommonCache.getNameServerClient().initConnection();
        CommonCache.getNameServerClient().sendRegistryMsg();
        CommonCache.getNameServerClient().setBrokerMasterChangeCallBack(new BrokerMasterChangeListener());
        // 判断是否是集群模式且是主节点，是主节点的话无需继续
        if (!BrokerClusterModeEnum.MASTER_SLAVE.getCode().equals(CommonCache.getGlobalProperties().getBrokerClusterMode())
                || "master".equals(CommonCache.getGlobalProperties().getBrokerClusterRole())) {
            return;
        }
        // 是从节点，向broker主节点发送同步请求
        String masterAddress = CommonCache.getNameServerClient().queryBrokerMasterAddress();
        if(masterAddress != null) {
            //尝试链接主broker
            slaveSyncService = new SlaveSyncService();
            CommonCache.setSlaveSyncService(slaveSyncService);
            boolean connectionStat = slaveSyncService.connectMasterBrokerNode(masterAddress);
            if(connectionStat) {
                slaveSyncService.sendStartSyncMsg();
            }
        }
    }

    private static void initReBalanceJob() {
        CommonCache.getConsumerInstancePool().startReBalanceJob();
    }

    private static void initBrokerServer() throws InterruptedException {
        BrokerServer brokerServer = new BrokerServer(CommonCache.getGlobalProperties().getBrokerPort());
        brokerServer.startServer();
    }

    public static void main(String[] args) throws Exception {
        initProperties();
        //testConsumeQueue();
        initNameServerChannel();
        initReBalanceJob();
        initBrokerServer();
    }

    void testCommitReadAndWrite() {
        //for (int i = 0; i < 100; i++) {
        //    commitLogAppendHandler.appendMsg("order_cancel_topic", ("hello world" + i).getBytes());
        //    TimeUnit.SECONDS.sleep(2);
        //    LOGGER.info("已发送消息: {}", i);
        //}
        //CommitLogMMapFileModel commitLogMMapFileModel = CommonCache.getCommitLogMMapFileModelManager().get("order_cancel_topic");
        //for (int i = 0, offset = 0; i < 100; i++) {
        //    int length = ("hello world" + i).getBytes().length;
        //    byte[] bytes = commitLogMMapFileModel.readContent(offset, length);
        //    LOGGER.info("读取到的消息: {}", new String(bytes));
        //    offset += length;
        //    TimeUnit.SECONDS.sleep(2);
        //}
    }

    //static void testConsumeQueue() {
    //    new Thread(() -> {
    //        try {
    //            for (int i = 0; i < 100; i++) {
    //                commitLogAppendHandler.appendMsg("order_cancel_topic", ("hello world" + i).getBytes());
    //                TimeUnit.SECONDS.sleep(1);
    //                if (i % 10 == 0) LOGGER.info("已发送消息: {}", i);
    //            }
    //        } catch (Exception e) {
    //            e.printStackTrace();
    //        }
    //    }).start();
    //    new Thread(() -> {
    //        try {
    //            for (int i = 0; i < 100; i++) {
    //                TimeUnit.SECONDS.sleep(8);
    //                ConsumeQueueConsumeReqModel reqModel = new ConsumeQueueConsumeReqModel("order_cancel_topic", "user_service_group", 0, 7);
    //                List<ConsumeMsgCommitLogDTO> consume = consumeQueueConsumeHandler.consume(reqModel);
    //                if (consume != null) {
    //                    consume.forEach(item -> LOGGER.info("消费到的消息: {}", new String(item.getBody())));
    //                    consumeQueueConsumeHandler.ack(new ConsumeQueueConsumeAckModel("order_cancel_topic", "user_service_group", 0, consume.size()));
    //                }
    //                TimeUnit.SECONDS.sleep(2);
    //            }
    //        } catch (Exception e) {
    //            e.printStackTrace();
    //        }
    //    }).start();
    //    new Thread(() -> {
    //        try {
    //            for (int i = 0; i < 100; i++) {
    //                TimeUnit.SECONDS.sleep(6);
    //                ConsumeQueueConsumeReqModel reqModel = new ConsumeQueueConsumeReqModel("order_cancel_topic", "order_service_group", 0, 5);
    //                List<ConsumeMsgCommitLogDTO> consume = consumeQueueConsumeHandler.consume(reqModel);
    //                if (consume != null) {
    //                    consume.forEach(item -> LOGGER.info("消费到的消息: {}", new String(item.getBody())));
    //                    consumeQueueConsumeHandler.ack(new ConsumeQueueConsumeAckModel("order_cancel_topic", "order_service_group", 0, consume.size()));
    //                }
    //
    //            }
    //        } catch (Exception e) {
    //            e.printStackTrace();
    //        }
    //    }).start();
    //}
}
